#include <cassert>
#include <iostream>

#include "metrics/JsonMetricFormatter.hpp"
#include "metrics/MetricRegistry.hpp"
#include "metrics/KafkaMetricReporter.hpp"

namespace metrics
{
    void DefaultKafkaDRCb::dr_cb(RdKafka::Message &message)
    {
        std::cout << "mesage deliver to topic: " << message.topic_name() << std::endl;
    }

    void DefaultKafkaEventCb::event_cb(RdKafka::Event &event)
    {
        std::cout << event.err() << ": " << event.str() << std::endl;
    }

    KafkaMetricReporter::KafkaMetricReporter(const std::string &brokers, const std::string &topic,
                                             std::unique_ptr<RdKafka::EventCb> event_cb,
                                             std::unique_ptr<RdKafka::DeliveryReportCb> dr_cb)
        : _formatter(new JsonMetricFormatter()), _brokers(brokers),
          _topic(topic), _event_cb(std::move(event_cb)), _dr_cb(std::move(dr_cb))
    {
        std::string errstr;
        auto res = _conf->set("bootstrap.servers", _brokers, errstr);
        if (res != RdKafka::Conf::CONF_OK)
        {
            std::cout << errstr << std::endl;
            exit(1);
        }
        if (_conf->set("dr_cb", _dr_cb.get(), errstr) != RdKafka::Conf::CONF_OK)
        {
            std::cerr << errstr << std::endl;
            exit(1);
        }
        if (_conf->set("event_cb", _event_cb.get(), errstr) != RdKafka::Conf::CONF_OK)
        {
            std::cerr << errstr << std::endl;
            exit(1);
        }
        _producer = RdKafka::Producer::create(_conf, errstr);
        if (!_producer)
        {
            std::cerr << "Failed to create producer: " << errstr << std::endl;
            exit(1);
        }
    }

    KafkaMetricReporter::~KafkaMetricReporter()
    {
        if (_formatter != nullptr)
        {
            delete _formatter;
        }
        _producer->flush(10 * 1000 /* wait for max 10 seconds */);
        if (_producer != nullptr)
        {
            delete _producer;
        }
        if (_conf != nullptr)
        {
            delete _conf;
        }
    }

    void KafkaMetricReporter::extra_metrics()
    {
    }

    void KafkaMetricReporter::run()
    {
        std::vector<metrics::MetricFamily> families = _registry->collect();
        for (auto family : families)
        {
            auto lines = _formatter->format(family);
            for (auto &line : lines)
            {
                _producer->produce(_topic, RdKafka::Topic::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY,
                                   const_cast<char *>(line.c_str()), line.size(), NULL, 0, 0, NULL);
            }

            // TODO: err
            _producer->poll(0);
        }
    }

} // namespace metrics
